package com.leyantech.ecp.agenx.task.consumer;

import com.leyantech.chaos.kafka.KafkaConsumerProcessor;
import com.leyantech.ecp.agenx.common.converter.Converter;
import com.leyantech.ecp.agenx.common.sender.Dispatcher;
import com.leyantech.ecp.agenx.common.utils.PbPrintHelper;
import com.leyantech.ecp.agenx.task.filter.TaskFilter;
import com.leyantech.utility.Logger;

import com.google.common.collect.ImmutableList;
import com.google.inject.Inject;
import com.google.protobuf.MessageOrBuilder;
import io.vertx.core.json.JsonObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-12-05.
 */
public abstract class TaskConsumer<Task> implements KafkaConsumerProcessor<String, Task> {

  protected Logger logger = new Logger(getClass());

  private final Dispatcher dispatcher;

  private final Converter<Task, JsonObject> converter;

  private final ImmutableList<TaskFilter<Task>> filters;

  @Inject
  protected TaskConsumer(
      Dispatcher dispatcher,
      Converter<Task, JsonObject> converter,
      ImmutableList<TaskFilter<Task>> filters) {
    this.dispatcher = dispatcher;
    this.converter = converter;
    this.filters = filters;
  }

  protected abstract String getStoreId(Task task);

  protected abstract String getAssistantId(Task task);

  Object taskLazyToString(Task task) {
    if (task instanceof MessageOrBuilder) {
      return PbPrintHelper.lazyToJson((MessageOrBuilder) task);
    }
    return task;
  }

  @Override
  public void accept(ConsumerRecord<String, Task> consumerRecord) {
    final Task task = consumerRecord.value();

    logger.info("received task {}", taskLazyToString(task));

    for (TaskFilter<Task> filter : filters) {
      if (!filter.test(task)) {
        return;
      }
    }

    final JsonObject taskMessage = converter.apply(task);

    final String storeId = getStoreId(task);
    final String assistantId = getAssistantId(task);

//    final String address = consumerRecord.key();

    dispatcher.sendToConnector(storeId, assistantId, taskMessage.toString());
  }

}
